package com.java.test.audit.push;

import com.java.test.audit.service.ExamineManageService;
import com.java.test.user.service.MemberService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
public class OrderExchangeWorker implements ApplicationRunner {
    @Autowired
    private ExamineManageService examineManageService;
    @Autowired
    private MemberService memberservice;
    @Value("${rocketMQ.nameSrvUrl}")
    private String nameSrvUrlSpring;
    @Value("${rocketMQ.returnProducerGroup}")
    private String returnProducerGroupSpring;
    @Value("${rocketMQ.returnNamespace}")
    private String returnNamespaceSpring;
    @Value("${rocketMQ.returnTopicClass}")
    private String returnTopicClassSpring;
    @Value("${rocketMQ.noticeProducerGroup}")
    private String noticeProducerGroupSpring;
    @Value("${rocketMQ.noticeNamespace}")
    private String noticeNamespaceSpring;
    @Value("${rocketMQ.noticeTopicClass}")
    private String noticeTopicClassSpring;
    @Value("${rocketMQ.user}")
    private String userSpring;
    @Value("${rocketMQ.password}")
    private String passwordSpring;
    private static String nameSrvUrl;
    private static String returnProducerGroup;              //返回给游戏平台
    private static String returnNamespace;                            //返回给游戏平台
    private static String returnTopicClass;                               //返回给游戏平台

    private static String noticeProducerGroup;           //游戏平台发送过来
    private static String noticeNamespace;                        //游戏平台发送过来
    private static String noticeTopicClass;                            //游戏平台发送给后台

    public static String user;
    public static String password;

	public static boolean sendChannelOrderState(String tagsName ,String orderId, String bodyStr) throws Exception{
        DefaultMQProducer producer = new DefaultMQProducer(OrderExchangeWorker.returnNamespace,OrderExchangeWorker.returnProducerGroup,new AclClientRPCHook(new SessionCredentials(OrderExchangeWorker.user,OrderExchangeWorker.password)));
        producer.setNamesrvAddr(OrderExchangeWorker.nameSrvUrl);
        producer.start();
        Message msg = new Message(OrderExchangeWorker.returnTopicClass, tagsName,orderId , bodyStr.getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.send(msg);
        System.out.println(sendResult.getSendStatus());
        boolean isOk=false;
        if(sendResult.getSendStatus().toString().equals("SEND_OK")){
            isOk=true;
        }
        producer.shutdown();
        return isOk;
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        nameSrvUrl = nameSrvUrlSpring;
        returnProducerGroup = returnProducerGroupSpring;
        returnNamespace = returnNamespaceSpring;
        returnTopicClass = returnTopicClassSpring;
        noticeProducerGroup = noticeProducerGroupSpring;
        noticeNamespace = noticeNamespaceSpring;
        noticeTopicClass = noticeTopicClassSpring;
        user = userSpring;
        password = passwordSpring;
        new GetOrderMsgWorkerThread(examineManageService,memberservice).start();
    }


    //主动获取
    public static class  GetOrderMsgWorkerThread extends Thread{
        ExamineManageService examineManageService1;
        MemberService memberservice1;
        //主动获取
        private DefaultLitePullConsumer pullConsumer;
        public GetOrderMsgWorkerThread (ExamineManageService examineManageService,MemberService memberservice) throws Exception{
            examineManageService1 = examineManageService;
            memberservice1 = memberservice;
            this.pullConsumer=new DefaultLitePullConsumer(OrderExchangeWorker.noticeNamespace,OrderExchangeWorker.noticeProducerGroup,new AclClientRPCHook(new SessionCredentials(OrderExchangeWorker.user,OrderExchangeWorker.password)));
            this.pullConsumer.setNamesrvAddr(nameSrvUrl);
            this.pullConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            //this.pullConsumer.setMessageModel(MessageModel.BROADCASTING);//广播
            this.pullConsumer.setMessageModel(MessageModel.CLUSTERING);//list
            this.pullConsumer.subscribe(OrderExchangeWorker.noticeTopicClass,"*");
            this.pullConsumer.start();
        }

        @Override
        public void run() {
            while (true){
                try {
                    //拉取消息，无消息时会阻塞
                    List<MessageExt> msgs = pullConsumer.poll();
                    for (MessageExt s:msgs){
                        try {
                            new TradeExamineResolving(examineManageService1,memberservice1).resolving(s);
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }
                    pullConsumer.commitSync();
                }catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }

   /* public static void main(String[] args) {

        try {
            while (true){
                int nowTime = (int) (System.currentTimeMillis() / 1000);
                System.out.println("开始发送" + nowTime);
                String orderId="d5e67c7b-53ee-42ad-b6bf-4f293c4bf6e8";
                JSONObject dataObject=new JSONObject();
                dataObject.put("code",1);
                JSONArray data=new JSONArray();
                data.put("09b561b2-9272-47ee-838f-d85aaf40a329");
                data.put(0);
                data.put(orderId);
                dataObject.put("data",data);
                OrderExchangeWorker.sendChannelOrderState("TagA_1",orderId,dataObject.toString());
                Thread.sleep(1000);
            }
        }catch (Exception r){r.printStackTrace();}

    }*/
}


